package com.xian.scala.jdbc

import org.apache.flink.api.common.typeinfo.BasicTypeInfo
import org.apache.flink.api.java.io.jdbc.JDBCInputFormat
import org.apache.flink.api.java.typeutils.RowTypeInfo
import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.api.scala._

/**
 * 获取mysql数据
 */
object GetMysqlData {
  def main(args: Array[String]): Unit = {
    val env = ExecutionEnvironment.getExecutionEnvironment

    //准备mysql相关信息



    val inputFormat = JDBCInputFormat.buildJDBCInputFormat()
      .setDrivername("com.mysql.jdbc.Driver")
      .setDBUrl("jdbc:mysql://r71/db1")
      .setUsername("root")
      .setPassword("123456")
      .setQuery("SELECT id,NAME,birthday FROM testdb.person")
      .setRowTypeInfo(new RowTypeInfo(
        BasicTypeInfo.INT_TYPE_INFO,
        BasicTypeInfo.STRING_TYPE_INFO,
        BasicTypeInfo.DATE_TYPE_INFO
      ))
      .finish()

    val value = env.createInput(inputFormat)

    value.print()


  }
}
